package com.atguigu.app.dwd;

import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class PaymentWideApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //1.1 设置状态后端
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/dwd_log/ck"));
        //1.2 开启CK
        //env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(60000L);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(15));

        //TODO 2.读取Kafka topic_db 和 dwd_order_wide 主题数据
        //2.1 读取Kafka topic_db主题数据
        tableEnv.executeSql("" +
                "create table max_well( " +
                "    `database` string, " +
                "    `table` string, " +
                "    `type` string, " +
                "    `data` map<string,string> " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "payment_wide_app_test"));

        //2.2 读取dwd_order_wide主题数据
        tableEnv.executeSql("" +
                "create table order_wide( " +
                "    order_detail_id string, " +
                "    order_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    order_price string, " +
                "    sku_num string, " +
                "    create_time string, " +
                "    source_type string, " +
                "    source_id string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    consignee string, " +
                "    consignee_tel string, " +
                "    total_amount string, " +
                "    order_status string, " +
                "    user_id string, " +
                "    payment_way string, " +
                "    delivery_address string, " +
                "    order_comment string, " +
                "    out_trade_no string, " +
                "    trade_body string, " +
                "    operate_time string, " +
                "    expire_time string, " +
                "    process_status string, " +
                "    tracking_no string, " +
                "    parent_order_id string, " +
                "    province_id string, " +
                "    activity_reduce_amount string, " +
                "    coupon_reduce_amount string, " +
                "    original_total_amount string, " +
                "    feight_fee string, " +
                "    feight_fee_reduce string, " +
                "    refundable_time string, " +
                "    order_activity_id string, " +
                "    activity_id string, " +
                "    activity_rule_id string, " +
                "    order_coupon_id string, " +
                "    coupon_id string, " +
                "    coupon_use_id string " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_order_wide", "payment_wide_app_test"));

        //TODO 3.过滤出支付表的数据
        Table paymentInfo = tableEnv.sqlQuery("" +
                "select " +
                "    data['id'] id, " +
                "    data['out_trade_no'] out_trade_no, " +
                "    data['order_id'] order_id, " +
                "    data['user_id'] user_id, " +
                "    data['payment_type'] payment_type, " +
                "    data['trade_no'] trade_no, " +
                "    data['total_amount'] total_amount, " +
                "    data['subject'] subject, " +
                "    data['payment_status'] payment_status, " +
                "    data['create_time'] create_time, " +
                "    data['callback_time'] callback_time, " +
                "    data['callback_content'] callback_content " +
                "from max_well " +
                "where `table` = 'payment_info' " +
                "and `database`='gmall-210927-flink'" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("payment_info", paymentInfo);
        //tableEnv.sqlQuery("select * from payment_info").execute().print();

        //TODO 4.关联两张表
        Table result = tableEnv.sqlQuery("" +
                "select " +
                "    ow.order_detail_id, " +
                "    ow.order_id, " +
                "    ow.sku_id, " +
                "    ow.sku_name, " +
                "    ow.order_price, " +
                "    ow.sku_num, " +
                "    ow.create_time, " +
                "    ow.source_type, " +
                "    ow.source_id, " +
                "    ow.split_total_amount, " +
                "    ow.split_activity_amount, " +
                "    ow.split_coupon_amount, " +
                "    ow.consignee, " +
                "    ow.consignee_tel, " +
                "    ow.total_amount, " +
                "    ow.order_status, " +
                "    ow.user_id, " +
                "    ow.payment_way, " +
                "    ow.delivery_address, " +
                "    ow.order_comment, " +
                "    ow.out_trade_no, " +
                "    ow.trade_body, " +
                "    ow.operate_time, " +
                "    ow.expire_time, " +
                "    ow.process_status, " +
                "    ow.tracking_no, " +
                "    ow.parent_order_id, " +
                "    ow.province_id, " +
                "    ow.activity_reduce_amount, " +
                "    ow.coupon_reduce_amount, " +
                "    ow.original_total_amount, " +
                "    ow.feight_fee, " +
                "    ow.feight_fee_reduce, " +
                "    ow.refundable_time, " +
                "    ow.order_activity_id, " +
                "    ow.activity_id, " +
                "    ow.activity_rule_id, " +
                "    ow.order_coupon_id, " +
                "    ow.coupon_id, " +
                "    ow.coupon_use_id, " +
                "    pi.id payment_id, " +
                "    pi.trade_no, " +
                "    pi.subject, " +
                "    pi.payment_status, " +
                "    pi.create_time payment_create_time, " +
                "    pi.callback_time, " +
                "    pi.callback_content " +
                "from payment_info pi " +
                "join order_wide ow " +
                "on pi.order_id = ow.order_id");
        tableEnv.createTemporaryView("result_table", result);
        //tableEnv.sqlQuery("select * from result_table").execute().print();

        //TODO 5.将数据写出
        tableEnv.executeSql("" +
                "create table payment_wide( " +
                "    order_detail_id string, " +
                "    order_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    order_price string, " +
                "    sku_num string, " +
                "    create_time string, " +
                "    source_type string, " +
                "    source_id string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    consignee string, " +
                "    consignee_tel string, " +
                "    total_amount string, " +
                "    order_status string, " +
                "    user_id string, " +
                "    payment_way string, " +
                "    delivery_address string, " +
                "    order_comment string, " +
                "    out_trade_no string, " +
                "    trade_body string, " +
                "    operate_time string, " +
                "    expire_time string, " +
                "    process_status string, " +
                "    tracking_no string, " +
                "    parent_order_id string, " +
                "    province_id string, " +
                "    activity_reduce_amount string, " +
                "    coupon_reduce_amount string, " +
                "    original_total_amount string, " +
                "    feight_fee string, " +
                "    feight_fee_reduce string, " +
                "    refundable_time string, " +
                "    order_activity_id string, " +
                "    activity_id string, " +
                "    activity_rule_id string, " +
                "    order_coupon_id string, " +
                "    coupon_id string, " +
                "    coupon_use_id string, " +
                "    payment_id string, " +
                "    trade_no string, " +
                "    subject string, " +
                "    payment_status string, " +
                "    payment_create_time string, " +
                "    callback_time string, " +
                "    callback_content string, " +
                "    PRIMARY KEY (payment_id) NOT ENFORCED " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_payment_wide"));
        tableEnv.executeSql("insert into payment_wide select * from result_table").print();

        //TODO 6.启动任务
        env.execute("PaymentWideApp");

    }

}
